
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <pthread.h>
#include <unistd.h>

#include "threadpool.h"

const int NUMBER = 2;

// 任务结构体
struct Task{
    void (*function)(void* arg);
    void* arg;
};

// 线程池结构体
struct ThreadPool{
    // 任务队列
    struct Task* taskQ;
    int QueueCapacity;      // 容量
    int queueSize;          // 当前任务个数
    int queueFront;         // 对头->取任务
    int queueRear;          // 对尾->放任务

    pthread_t managetID;    // 管理的线程ID
    pthread_t *threadIDs;   // 工作的线程ID
    int maxNum;             // 最大线程数量
    int minNum;             // 最小线程的数量
    int busyNum;            // 忙的线程个数
    int liveNum;            // 存活的线程个数
    int exitNum;            // 要销毁的线程个数
    pthread_mutex_t mutexPool;// 锁整个任务队列
    pthread_mutex_t mutexBusy;// 锁busyNum变量
    pthread_cond_t notFull;     // 任务队列是不是满了
    pthread_cond_t notEmpty;    // 任务队列是不是空了

    int shutdown;               // 是不是要销毁线程池，1 销毁，0 不销毁
};

struct ThreadPool *threadPoolCreate(int min, int max, int queueSize)
{
    struct ThreadPool *pool= (struct ThreadPool*)malloc(sizeof(struct ThreadPool));
    do
    {
        if(NULL == pool)
        {
            printf("malloc threadpool fail...\n");
            break;
        }
    
        pool->threadIDs = (pthread_t*)malloc(sizeof(pthread_t) * max);
        if(NULL == pool->threadIDs)
        {
            printf("malloc threadIDs fail...\n");
            break;
        }
    
        memset(pool->threadIDs, 0, sizeof(pthread_t) * max);
        pool->maxNum = max;
        pool->minNum = min;
        pool->busyNum = 0;
        pool->liveNum = min;
        pool->exitNum = 0;
    
        if(pthread_mutex_init(&pool->mutexPool, NULL) != 0 ||
            pthread_mutex_init(&pool->mutexBusy, NULL) != 0 ||
            pthread_cond_init(&pool->notFull, NULL) != 0 ||
            pthread_cond_init(&pool->notEmpty, NULL) != 0)
        {
            printf("mutex or cond fail...\n");
            break;
        }
    
        // 任务队列
        pool->taskQ = (struct Task*)malloc(sizeof(struct Task) * queueSize);
        pool->QueueCapacity = queueSize;
        pool->queueSize = 0;
        pool->queueFront = 0;
        pool->queueRear = 0;
    
        pool->shutdown = 0;
    
        // 创建线程
        pthread_create(&pool->managetID, NULL, managet, pool);

        for(int i = 0; i < min; ++i)
        {
            pthread_create(&pool->threadIDs[i], NULL, worker, pool);
        }

        return pool;

    }while(0);

    // 释放资源
    if(pool && pool->threadIDs) free(pool->threadIDs);
    if(pool && pool->taskQ) free(pool->taskQ);
    if(pool) free(pool);

    return NULL;
}

int threadPoolDestroy(struct ThreadPool* pool)
{
    if(NULL == pool)
    {
        return -1;
    }
    pool->shutdown = 1;
    // 阻塞回收管理者线程
    pthread_join(pool->managetID, NULL);
    // 唤醒阻塞的消费者线程
    for(int i = 0; i < pool->liveNum; ++i)
    {
        pthread_cond_signal(&pool->notEmpty);
    }
    // 释放堆内存
    if(pool->taskQ)
    {
        free(pool->taskQ);
    }
    if(pool->threadIDs)
    {
        free(pool->threadIDs);
    }
    
    pthread_mutex_destroy(&pool->mutexPool);
    pthread_mutex_destroy(&pool->mutexBusy);
    pthread_cond_destroy(&pool->notEmpty);
    pthread_cond_destroy(&pool->notFull);
    free(pool);
    pool = NULL;

    return 0;
}

void threadPoolAdd(struct ThreadPool* pool, void(*func)(void*), void* arg)
{
    pthread_mutex_lock(&pool->mutexPool);
    while(pool->queueSize == pool->QueueCapacity && !pool->shutdown)
    {
        //阻塞生产者线程
        pthread_cond_wait(&pool->notFull, &pool->mutexPool);
    }
    if(pool->shutdown)
    {
        pthread_mutex_unlock(&pool->mutexPool);
        return;
    }
    //添加任务
    pool->taskQ[pool->queueRear].function = func;
    pool->taskQ[pool->queueRear].arg = arg;
    pool->queueRear = (pool->queueRear + 1) % pool->QueueCapacity;
    pool->queueSize++;

    pthread_cond_signal(&pool->notEmpty);
    pthread_mutex_unlock(&pool->mutexPool);
}

int threadPoolBusyNumbber(struct ThreadPool* pool)
{
    pthread_mutex_lock(&pool->mutexBusy);
    int busyNum = pool->busyNum;
    pthread_mutex_unlock(&pool->mutexBusy);
    return busyNum;
}

int threadPoolAliveNumbber(struct ThreadPool* pool)
{
    pthread_mutex_lock(&pool->mutexPool);
    int aliveNum = pool->liveNum;
    pthread_mutex_unlock(&pool->mutexPool);
    return aliveNum;
}

void* worker(void* arg)
{
    struct ThreadPool* pool =(struct ThreadPool*)arg;
    while(1)
    {
        pthread_mutex_lock(&pool->mutexPool);
        // 任务队列是否为空
        while(0 == pool->queueSize && !pool->shutdown)
        {
            // 阻塞工作线程
            pthread_cond_wait(&pool->notEmpty, &pool->mutexPool);

            // 判断是不是要销毁线程
            if(pool->exitNum > 0)
            {
                pool->exitNum--;
                if(pool->liveNum > pool->minNum)
                {
                    pool->liveNum--;
                    pthread_mutex_unlock(&pool->mutexPool);
                    threadExit(pool);
                }
                
            }
        }

        // 判断线程是否关闭
        if(pool->shutdown)
        {
            pthread_mutex_unlock(&pool->mutexPool);
            threadExit(pool);
        }

        // 从任务队列取出任务
        struct Task task;
        task.function = pool->taskQ[pool->queueFront].function;
        task.arg = pool->taskQ[pool->queueFront].arg;
        // 移动头结点
        pool->queueFront = (pool->queueFront + 1) % pool->QueueCapacity;
        pool->queueSize--;
        // 解锁
        pthread_cond_signal(&pool->notFull);
        pthread_mutex_unlock(&pool->mutexPool);

        printf("thread %ld start working...\n", pthread_self());
        pthread_mutex_lock(&pool->mutexBusy);
        pool->busyNum++;
        pthread_mutex_unlock(&pool->mutexBusy);

        task.function(task.arg);
        free(task.arg);
        task.arg = NULL;

        printf("thread %ld end working...\n", pthread_self());
        pthread_mutex_lock(&pool->mutexBusy);
        pool->busyNum--;
        pthread_mutex_unlock(&pool->mutexBusy);

    }
}

void* managet(void* arg)
{
    struct ThreadPool* pool =(struct ThreadPool*)arg;
    while(!pool->shutdown)
    {
        // 每隔3秒
        sleep(3);
        // 取出线程池中任务的数量和当前线程的数量
        pthread_mutex_lock(&pool->mutexPool);
        int queueSize = pool->queueSize;
        int liveNum = pool->liveNum;
        pthread_mutex_unlock(&pool->mutexPool);

        // 取出忙的工作线程数量
        pthread_mutex_lock(&pool->mutexBusy);
        int busyNum = pool->busyNum;
        pthread_mutex_unlock(&pool->mutexBusy);

        // 添加线程
        // 任务个数>存活的线程个数 && 存活的线程个数<最大线程个数
        if(queueSize > liveNum && liveNum < pool->maxNum)
        {
            pthread_mutex_lock(&pool->mutexPool);
            int counter = 0;
            for(int i =0; i < pool->maxNum && counter < NUMBER && pool->liveNum < pool->maxNum; ++i)
            {
                if(pool->threadIDs[i] == 0)
                {
                    pthread_create(&pool->threadIDs[i], NULL, worker, pool);
                    counter++;
                    pool->liveNum++;
                }
                
            }
            pthread_mutex_unlock(&pool->mutexPool);

        }

        // 销毁线程
        // 忙的线程*2 < 存活的线程数 && 存活的线程数 > 最小线程数
        if(busyNum*2 < liveNum && liveNum > pool->minNum)
        {
            pthread_mutex_lock(&pool->mutexPool);
            pool->exitNum = NUMBER;
            pthread_mutex_unlock(&pool->mutexPool);

            // 让工作的线程自杀
            for(int i = 0; i < NUMBER; ++i)
            {
                pthread_cond_signal(&pool->notEmpty);
            }
        }

    }
}

void threadExit(struct ThreadPool* pool)
{
    pthread_t tid = pthread_self();
    for(int i = 0; i < pool->maxNum; ++i)
    {
        if(pool->threadIDs[i] == tid)
        {
            pool->threadIDs[i] = 0;
            printf("thread Exit called, %ld exiting...\n", tid);
            break;
        }
    }
    pthread_exit(NULL);
}

